package club.monkeywood.ad.dmp.report

import club.monkeywood.ad.dmp.bean.Log
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

/**
	* 统计日志文件中省市的数据量分布情况
	*
	*  本次统计是基于parquet文件
	*
	* 需求1：
	*     将统计出来的结果存储成json文件格式
	*
	* 需求2：
	*     将统计出来的结果存储到mysql中
	*/
object ProCityRptBySql {

	def main(args: Array[String]): Unit = {

		// 0 校验参数个数
		if (args.length != 2) {
			println(
				"""
					|club.monkeywood.ad.dmp.report.ProCityRpt
					|参数：
					| logInputPath
					| resultOutputPath
				""".stripMargin)
			sys.exit()
		}

		// 1 接受程序参数
		val Array(logInputPath, resultOutputPath) = args

		// 2 创建sparkconf->SparkSession
		val sparkConf = new SparkConf()
		sparkConf.setAppName(s"${this.getClass.getSimpleName}")
		sparkConf.setMaster("local[*]")
		//使用KryoSerializer更快
		// RDD 序列化到磁盘 worker与worker之间的数据传输
		sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

		val ss = SparkSession.builder()
			.appName(s"${this.getClass.getSimpleName}")
			.master("local[*]")
			.config(sparkConf)
			.getOrCreate()

		// 读取数据 --》parquet文件
		val df: DataFrame = ss.read.parquet(logInputPath)

		// 将dataframe注册成一张临时表
		df.createTempView("log")

		df.show()

		// 按照省市进行分组聚合---》统计分组后的各省市的日志记录条数
		val result: DataFrame = ss.sql("select provincename, cityname, count(*) ct from log group by provincename, cityname")

		// 判断结果存储路径是否存在，如果存在则删除
		// core-site.xml中的fs.defaultFS默认值：file://：表示本地磁盘路径，所以输出文件会写入本地路径
		// 若改为hdfs路径，则写入hdfs中
		val hadoopConfiguration = ss.sparkContext.hadoopConfiguration
		val fs = FileSystem.get(hadoopConfiguration)
		val resultPath = new Path(resultOutputPath)
		if(fs.exists(resultPath)) {
			fs.delete(resultPath, true)
		}

		// 将结果存储的成json文件
		//result.write.json(resultOutputPath)
		// 合并分区
		result.coalesce(1)
			.write
			.json(resultOutputPath)

		ss.stop()
	}

}
